/*线程池内容：
 *  1.一个线程安全的资源队列（元素：数据+处理方法）
 *  2.固定数量的处理线程
 * */

#include <stdio.h>
#include <queue>
#include <pthread.h>

#define THREAD_POOL_SIZE 20
#define THREAS_POOL_THREAD 5 
#define PRODUCTION_COUNT 1000

using namespace std;

/* 线程池资源队列的元素类
 * */
// 定义一个函数指针类型
typedef void(*Handler)(int data);

class QueueData
{
    public:
        QueueData(){};  //无参构造
        QueueData(int data, Handler handler)
        {
            _data = data;
            _handler = handler;
        }
        void run()  //执行元素相应操作
        {
            _handler(_data);
        }
        

    private:
        int _data;  //数据
        Handler _handler;  //该数据对应的处理函数（由生产者决定）
};

/* 封装一个处理 生产者-消费者模型 的线程类
 * */
class ThreadPool
{
    public:
        ThreadPool(int size, int thread_count )
            : _capacity(size)
            , _thread_count(thread_count)
            , _flag_exit(0)
        {
            pthread_mutex_init(&_que_lock, NULL);
            pthread_cond_init(&_cons_cond, NULL);
            pthread_cond_init(&_prod_cond, NULL);
        }
        ~ThreadPool()
        {
            pthread_mutex_destroy(&_que_lock);
            pthread_cond_destroy(&_cons_cond);
            pthread_cond_destroy(&_prod_cond);
        }
        // 资源队列上锁
        void QueLock()
        {
            pthread_mutex_lock(&_que_lock);
        }
        // 资源队列解锁
        void QueUnlock()
        {
            pthread_mutex_unlock(&_que_lock);
        }

        // 消费线程的Pop接口
        void Pop(QueueData* qd)
        {
            *qd = _Que.front();
            _Que.pop();
        }

        // 生产线程的Push接口
        void Push(QueueData qd)
        {
            // Push接口内保证队列线程安全和同步
            QueLock();
            // 判断线程池是否退出
            if(_flag_exit)
            {
                QueUnlock();
                return;
            }
            while(_Que.size() >= _capacity)
            {
                pthread_cond_wait(&_prod_cond, &_que_lock);
            }
            _Que.push(qd);
            QueUnlock();
            pthread_cond_signal(&_cons_cond);
        }
        
        // 线程池（消费）线程的入口函数
        // 线程的启动函数参数只能有一个且必须为void*类型
        // 为了避免类成员函数默认参数this指针的影响，用static修饰为静态成员函数
        static void* thread_poll_start(void* arg)
        {
            // 让线程池线程设置分离属性，防止僵尸线程（线程池线程不使用线程等待）
            pthread_detach(pthread_self());
            // 参数传入线程池
            ThreadPool* tp = (ThreadPool*)arg;
            // 循环读取线程队列元素，并执行响应的操作
            while(1)
            {
                // 取锁
                tp->QueLock();
                // 若资源队列无元素
                while(tp->_Que.empty())
                {
                    // 先判断线程池调用是否结束
                    if(tp->_flag_exit)
                    {
                        // 若结束，则exit该线程
                        tp->_thread_count--;
                        tp->QueUnlock();  //任何线程退出的地方，记得解锁
                        pthread_exit(NULL);
                    }
                    // 若线程池仍运行，该线程进行等待
                    pthread_cond_wait(&tp->_cons_cond, &tp->_que_lock);
                }
                
                // 读取元素
                QueueData qd;
                tp->Pop(&qd);

                // 先解锁再执行元素操作，防止影响其他线程获取队列资源
                tp->QueUnlock();
                pthread_cond_signal(&tp->_prod_cond);

                // 执行元素接口
                qd.run();
            }
        }
        
        // 创建消费线程（消费线程的创建直接在线程池类中定义）
        int OnInit()
        {
            int cur = 0;  //线程创建失败的个数
            int i = 0;
            for(; i < _thread_count; ++i)
            {
                pthread_t tid;  //线程标志无需保留，因为线程池线程调用exit接口退出，无需线程等待
                int ret = pthread_create(&tid, NULL, thread_poll_start, (void*)this);
                if(ret < 0)
                {
                    cur++;
                }
            }
            return _thread_count - cur;  //返回成功创建线程的个数
        }

        // 线程池退出
        void ThreadPool_exit()
        {
            _flag_exit = 1;  //修改此标志位无需加锁，即使多线程执行，也不会造成二义性结果
            while(_thread_count > 0)  //此时还可能存在线程等待的线程，需唤醒才能执行exit
            {
                pthread_cond_signal(&_cons_cond);
                pthread_cond_signal(&_prod_cond);
            }
        }
        

    private:
        queue<QueueData> _Que;  //资源队列
        size_t _capacity;  //资源队列大小
        size_t _thread_count;  //线程池线程数量（消费者线程）
        int _flag_exit;  //线程池退出标志位
        
        pthread_mutex_t _que_lock;  //资源队列的互斥锁
        pthread_cond_t _cons_cond;  //消费线程的条件变量
        pthread_cond_t _prod_cond;  //生产线程的条件变量
};

// 数据的处理函数
void Data_Fun(int data)
{
    printf("data = %d\n", data);
}

int main()
{
    // 创建线程池
    ThreadPool* tp = new ThreadPool(THREAD_POOL_SIZE, THREAS_POOL_THREAD);
    
    // 启动线程池（启动消费者线程）
    if(tp->OnInit() <= 0)
    {
        printf("ThreadPool OnInit Wrong]\n");
        return 0;
    }

    // 主线程执行生产线程功能
    int i = 0;
    for(; i < PRODUCTION_COUNT; ++i)
    {
        QueueData qd(i, Data_Fun);
        tp->Push(qd);
    }

    // 线程池退出
    tp->ThreadPool_exit();
        
    delete tp;
    return 0;
}
